package avicit.bdp.dcs.datasource.service;

import avicit.bdp.common.datasource.BaseDataSource;
import avicit.bdp.common.datasource.HdfsDataSource;
import avicit.bdp.common.datasource.HiveDataSource;
import avicit.bdp.common.dto.BdpPrmEngineResourceDTO;
import avicit.bdp.common.dto.StorageResourceConf;
import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.service.service.DataSourceService;
import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.common.utils.DESUtils;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.datasource.service.ds.BaseDS;
import avicit.bdp.dcs.datasource.service.ds.DSFactory;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.utils.JdbcConstants;
import avicit.bdp.dms.api.BdpPrmEngineResourceApi;
import avicit.platform6.core.exception.BusinessException;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @金航数码科技有限责任公司
 * @作者：liyb
 * @邮箱：liyb@avic-digital.com
 * @创建时间： 2021-01-28 14:24
 * @类说明：数据源检测工具类
 * @修改记录：
 */
@Component
public class DataSourceUtils {

    private static final Logger logger = LoggerFactory.getLogger(DataSourceUtils.class);

    @Autowired
    private DataSourceService dataSourceService;
    @Autowired
    private BaseDbService baseDbService;
    @Autowired
    private BdpPrmEngineResourceApi bdpPrmEngineResourceApi;

    /**
     * 将ds数据源转换成 -> JobDatasource
     */
    public JobDatasource convertDatasource(String datasourceId, String systemBuilt) throws Exception {
        // 系统内置数据源
        if (StringUtils.isNotEmpty(systemBuilt) && "1".equals(systemBuilt)) {
            BdpPrmEngineResourceDTO engineResourceDTO =
                    bdpPrmEngineResourceApi.getEngineResourceById(datasourceId);
            JobDatasource jobDatasource = new JobDatasource();

            if (engineResourceDTO != null) {
                StorageResourceConf conf = engineResourceDTO.getStorageResourceConf();
                jobDatasource.setId(datasourceId);
                jobDatasource.setSystemBuilt(1);
                int resourceType = engineResourceDTO.getResourceType();
                if (resourceType == 2) {
                    jobDatasource.setDbType(JdbcConstants.HDFS);
                    jobDatasource.setJdbcUrl(conf.getDeFaultFs());
                    jobDatasource.setKerberos("kerberos".equalsIgnoreCase(conf.getAuthType()));
                    jobDatasource.setStorageResourceConf(conf);
                } else if (resourceType == 3) {
                    jobDatasource.setDbType(JdbcConstants.OSS);
                    jobDatasource.setEndPoint(conf.getEndPoint());
                    jobDatasource.setAccessKey(conf.getAccessKey());
                    jobDatasource.setSecretKey(conf.getSecretKey());
                    jobDatasource.setBucket(conf.getStoragePath());
                }
            }
            return jobDatasource;
        } else {
            JobDatasource jobDatasource = convertDatasource(datasourceId);
            jobDatasource.setSystemBuilt(0);
            return jobDatasource;
        }
    }

    /**
     * 将ds数据源转换成 -> JobDatasource
     */
    public JobDatasource convertDatasource(String datasourceId) throws Exception {
        DataSourceDTO dataSourceDTO = dataSourceService.queryDataSourceById(datasourceId);
        return convertDatasource(dataSourceDTO);
    }

    private JobDatasource convertDatasource(DataSourceDTO dataSourceDTO) throws Exception {
        JobDatasource jobDatasource = new JobDatasource();
        jobDatasource.setDataSourceDTO(dataSourceDTO);
        String datasourceType = dataSourceDTO.getDatasourceType();
        BaseDataSource baseDataSource = DBUtils.getBaseDataSource(dataSourceDTO);
        jobDatasource.setBaseDataSource(baseDataSource);
        String jdbcUrl = baseDataSource.getJdbcUrl();

        BaseDS baseDS = DSFactory.getDatasource(dataSourceDTO);
        if (baseDS == null) {
            String errMsg = "错误的datasourceType:" + datasourceType;
            logger.error(errMsg);
            throw new BusinessException(errMsg);
        }

        CheckParaUtils.checkStringValidity(jdbcUrl, logger, "数据源jdbcUrl属性没设置");

        handleUserAndPwd(baseDS, baseDataSource, jobDatasource);

        jobDatasource.setDbType(baseDS.getDatasource());
        jobDatasource.setJdbcDriverClass(baseDS.getJdbcDriverClass());

        jobDatasource.setJdbcUrl(handleDatasourceUrl(datasourceType, baseDataSource));
        jobDatasource.setId(dataSourceDTO.getId());

        String currentSchema = baseDbService.getCurrentSchema(dataSourceDTO.getId());
        jobDatasource.setDatabaseName(currentSchema);

        // hdfs设置kerberos
        if (JdbcConstants.HDFS.equals(datasourceType)) {
            HdfsDataSource hdfsDataSource = (HdfsDataSource) baseDataSource;
            if (hdfsDataSource.getKerberos() == 1) {
                jobDatasource.setKerberos(true);
            } else {
                jobDatasource.setKerberos(false);
            }
        }
        return jobDatasource;
    }

    private void handleUserAndPwd(
            BaseDS baseDS, BaseDataSource baseDataSource, JobDatasource jobDatasource) {
        String jdbcUsername = baseDataSource.getUser();
        String jdbcPassword = baseDataSource.getPassword();

        // hive开启kerberos认证，则不验证账号密码
        if (JdbcConstants.HIVE.equals(baseDS.getDatasource())) {
            HiveDataSource hiveDataSource = (HiveDataSource) baseDataSource;
            if (hiveDataSource.getKerberos() == 1) {
                return;
            }
        }

        if (baseDS.needUserAndPassword()) {
            CheckParaUtils.checkStringValidity(jdbcUsername, logger, "数据源jdbcUsername属性没设置");
            CheckParaUtils.checkStringValidity(jdbcPassword, logger, "数据源jdbcPassword属性没设置");

            jobDatasource.setJdbcUsername(jdbcUsername);
            jobDatasource.setEncryptJdbcPassword(jdbcPassword);
            // 需要对密码进行解密
            jobDatasource.setJdbcPassword(DESUtils.decrypt(jdbcPassword));
        }
    }

    /**
     * 处理jdbcUrl 由于greenplum插件不支持此格式jdbc:pivotal:greenplum://192.168.0.35:5432;databaseName=dbtest
     *
     * @param datasourceType
     * @param baseDataSource
     * @return
     */
    private String handleDatasourceUrl(String datasourceType, BaseDataSource baseDataSource) {
        String newUrl = baseDataSource.getJdbcUrl();
        if (JdbcConstants.GREENPLUM.equals(datasourceType)) {
            // jdbc:postgresql://192.168.0.35:5432/dbtest?currentSchema=myschema
            newUrl =
                    "jdbc:postgresql://"
                            + baseDataSource.getIp()
                            + ":"
                            + baseDataSource.getPort()
                            + "/"
                            + baseDataSource.getDatabase()
                            + "?"
                            + baseDataSource.getOther();
        }
        return newUrl;
    }

    /**
     * 验证源数据源和目标数据源是否相同
     *
     * @param srcDatasourceId
     * @param dstDatasourceId
     * @return
     */
    public boolean checkSrcAndDstDatasource(String srcDatasourceId, String dstDatasourceId) {
        DataSourceDTO srcDatasource = dataSourceService.queryDataSourceById(srcDatasourceId);
        BaseDataSource srcBaseDataSource = DBUtils.getBaseDataSource(srcDatasource);
        String srcDatasourceKey =
                srcBaseDataSource.getJdbcUrl()
                        + srcBaseDataSource.getUser()
                        + srcBaseDataSource.getPassword();

        DataSourceDTO dstDatasource = dataSourceService.queryDataSourceById(dstDatasourceId);
        BaseDataSource dstBaseDataSource = DBUtils.getBaseDataSource(dstDatasource);
        String dstDatasourceKey =
                dstBaseDataSource.getJdbcUrl()
                        + dstBaseDataSource.getUser()
                        + dstBaseDataSource.getPassword();

        return srcDatasourceKey.equals(dstDatasourceKey);
    }

    /**
     * 根据datasourceId获取数据源
     *
     * @param datasourceId
     * @return
     */
    public DataSourceDTO getDatasource(String datasourceId) throws Exception {
        DataSourceDTO dataSourceDTO = dataSourceService.queryDataSourceById(datasourceId);
        return dataSourceDTO;
    }

    /**
     * 根据datasourceId获取存储资源信息
     *
     * @param datasourceId
     * @return
     * @throws Exception
     */
    public StorageResourceConf getStorageResourceByDatasourceId(String datasourceId)
            throws Exception {
        DataSourceDTO dataSourceDTO = dataSourceService.queryDataSourceById(datasourceId);
        BaseDataSource baseDataSource = DBUtils.getBaseDataSource(dataSourceDTO);

        StorageResourceConf conf = new StorageResourceConf();
        conf.setId(dataSourceDTO.getId());
        conf.setDatasource(true);

        String dbType = dataSourceDTO.getDatasourceType().toLowerCase();
        if (JdbcConstants.HDFS.equals(dbType)) {
            conf.setType(2);
            HdfsDataSource hdfsDataSource = (HdfsDataSource) baseDataSource;
            conf.setDeFaultFs(hdfsDataSource.getAddress());
            conf.setStoragePath(hdfsDataSource.getFilePath());
            conf.setExecuteUser(hdfsDataSource.getHdfsUser());
            if (hdfsDataSource.getKerberos() == 1) {
                conf.setAuthType(Constants.KERBEROS);
                conf.setKrb5Path(hdfsDataSource.getKrb5Path());
                conf.setKeytabPath(hdfsDataSource.getKeytabPath());
            }

        } else if (JdbcConstants.OSS.equals(dbType)) {
            conf.setType(3);
        }

        return conf;
    }

    /**
     * 查询数据源详情
     *
     * @param jobDTO
     */
    public DataSourceDTO queryDataSourceById(JobDTO jobDTO) throws Exception {
        String datasourceId = jobDTO.getSrcDatasourceId();
        if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
            String systemBuilt = jobDTO.getExtraParams();

            // 系统内置数据源
            if (StringUtils.isNotEmpty(systemBuilt) && "1".equals(systemBuilt)) {
                BdpPrmEngineResourceDTO engineResourceDTO =
                        bdpPrmEngineResourceApi.getEngineResourceById(datasourceId);
                DataSourceDTO dataSource = new DataSourceDTO();
                if (engineResourceDTO != null) {
                    StorageResourceConf conf = engineResourceDTO.getStorageResourceConf();

                    dataSource.setId(datasourceId);
                    dataSource.setDatasourceName(engineResourceDTO.getName());
                    int resourceType = engineResourceDTO.getResourceType();
                    if (resourceType == 2) {
                        dataSource.setDatasourceType(JdbcConstants.HDFS);
                    } else if (resourceType == 3) {
                        dataSource.setDatasourceType(JdbcConstants.OSS);
                    }
                }
                return dataSource;
            } else {
                return dataSourceService.queryDataSourceById(datasourceId);
            }
        } else {
            return dataSourceService.queryDataSourceById(datasourceId);
        }
    }

    /**
     * 获取数据库Schema
     * @return
     * @throws Exception
     */
    public String getDbSchema(JobDatasource jobDatasource) throws Exception {
        String schema = "";
        if (Constants.DB_TYPE_ORACLE.equals(jobDatasource.getDbType())) {
            // 处理当前连接的schema
            BaseDataSource dbSource = jobDatasource.getBaseDataSource();
            String others = dbSource.getOther();
            if (StringUtils.isNotBlank(others) && others.contains("schema")) {
                schema = others.split("=")[1];
            }
        }

        return schema;
    }

    /**
     * 将schema和表名拼接在一起，schema.tableName
     * @param dbType
     * @param tableName
     * @param schema
     * @return
     */
    public static String appendSchema2TableName(String dbType, String tableName, String schema) {
        switch (dbType) {
            case Constants.DB_TYPE_ORACLE:
                if(StringUtils.isNotEmpty(schema)){
                    return "\"" + schema.toUpperCase() + "\"." + tableName;
                }
                return tableName;
            default:
                return tableName;
        }
    }

}
